package com.fs.iot.collector;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.fs.common.ErrorCodes;
import com.fs.common.config.NoticeProcessor;
import com.fs.common.exception.AppException;
import com.fs.iot.bean.dto.GatewayDTO;
import com.fs.iot.service.GatewayEntityService;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;

import java.util.*;

/**
 * 原始工况处理适配器
 *
 * @author LiuQi 2025/3/13-12:54
 * @version V1.0
 **/
@Service
@Slf4j
public class DataCollectAdapter implements NoticeProcessor {
    @Autowired
    private Set<AbstractDataCollector> dataCollectors;

    @Autowired
    @Lazy
    private GatewayEntityService gatewayEntityService;

    private final static Map<String, AbstractDataCollector> collectorMap = new HashMap<>(16);

    @PostConstruct
    public void init() {
        dataCollectors.forEach(collector -> collectorMap.put(collector.getProtocol(), collector));

        // 系统重启时需要启动监听
        List<GatewayDTO> list = gatewayEntityService.findAll();
        list.forEach(gateway -> {
            try {
                this.startGateway(gateway);
            } catch (Exception ex) {
                log.error("启动任务失败", ex);
            }
        });

        log.info("MQ任务自动启动完成");
    }

    @Override
    public void process(ConsumerRecord<String, String> record) {
        String key = record.key();
        if (!"gateway".equals(key)) {
            return;
        }

        String value = record.value();
        JSONObject json = JSON.parseObject(value);

        String gatewayId = json.getString("gatewayId");
        String op = json.getString("op");
        switch (op) {
            case "start" -> this.startGateway(gatewayId);
            case "stop" -> this.stopGateway(gatewayId);
            case "restart" -> {
                this.stopGateway(gatewayId);
                this.startGateway(gatewayId);
            }
        }
    }

    /**
     * 启动网关
     * @param gateway 网关
     */
    private void startGateway(GatewayDTO gateway) {
        String protocol = gateway.getProtocol();

        Optional.ofNullable(collectorMap.get(protocol)).ifPresent(collector -> {
            try {
                collector.startListener(gateway);
            } catch (Exception e) {
                log.error("网关启动失败", e);
                throw new RuntimeException(e);
            }
        });
    }

    /**
     * 启动网关监听
     */
    public void startGateway(String gatewayId) {
        GatewayDTO gateway = gatewayEntityService.findById(gatewayId).orElseThrow(AppException.supplier(ErrorCodes.IOT_GATEWAY_NOT_EXISTS));
        this.startGateway(gateway);
    }

    /**
     * 停止网关监听
     */
    public void stopGateway(String gatewayId) {
        GatewayDTO gateway = gatewayEntityService.findById(gatewayId).orElseThrow(AppException.supplier(ErrorCodes.IOT_GATEWAY_NOT_EXISTS));
        String protocol = gateway.getProtocol();
        Optional.ofNullable(collectorMap.get(protocol)).ifPresent(collector -> {
            collector.stopListener(gatewayId);
        });
    }
}
